package com.atguigu.gmall.realtime.sink;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.util.DruidDSUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author lzc
 * @Date 2022/12/3 14:00
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private DruidPooledConnection conn;
    private Jedis redisClient;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        conn = DruidDSUtil.getPhoenixConn();
        
        redisClient = RedisUtil.getRedisClient();
    }
    
    @Override
    public void close() throws Exception {
        if (conn != null) {
            // 不是关闭连接, 而是归还连接
            conn.close();
        }
        
        if (redisClient != null) {
            redisClient.close();
        }
    }
    
    // 当流中, 每来一个数据这个方法就执行一次
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> t,
                       Context context) throws Exception {
        JSONObject dim = t.f0;
        TableProcess tp = t.f1;
        
        // 1. 写维度到 phoenix
        writeDimToPhoenix(dim, tp);
        // 2. 更新或者删除维度
        // 更新: 先判断缓存中是有这个维度, 如果有再更新
        // 删除: 直接删除. (不存在没有任何的反应)
        delCacheFromRedis(dim, tp);
        
    }
    
    private void delCacheFromRedis(JSONObject dim, TableProcess tp) {
        // 如果这个维度是新增, 缓存中一定没有.
        // 当dim这次是 update 的时候,才有必要去删一次
        if ("update".equals(dim.getString("op_type"))) {
            System.out.println("删除缓存....");
            // 根据 sink 的表名和 id 组成一个 key
            String key = tp.getSinkTable() + ":" + dim.getString("id");
            // 删除指定的 key
            redisClient.del(key);
        }
        
    }
    
    private void writeDimToPhoenix(JSONObject dim, TableProcess tp) throws SQLException {
        // 使用标准的 jdbc 把数据写入到 phoenix 的表中
        
        // 1. 拼接一个 sql 语句
        // upsert into user(a,b,c)values(?,?,?)
        StringBuilder sql = new StringBuilder();
        sql
            .append("upsert into ")
            .append(tp.getSinkTable())
            .append("(")
            .append(tp.getSinkColumns())
            .append(")values(")
            .append(tp.getSinkColumns().replaceAll("[^,]+", "?"))
            .append(")");
        System.out.println("插入语句: " + sql);
        // 2. 使用 conn 对象获取一个预处理语句
        PreparedStatement ps = conn.prepareStatement(sql.toString());
        // 3.给 sql 中的占位符赋值
        String[] columns = tp.getSinkColumns().split(",");
        for (int i = 0; i < columns.length; i++) {
            String columnName = columns[i];
            String value = dim.getString(columnName);
            ps.setString(i + 1, value);
        }
        // 4. 执行
        ps.execute();
        // mysql会自动提交, phoenix 连接不会自动提交
        // 但是我们用的是连接池, 连接池设置的自动提交
        //conn.commit();
        // 5. 关闭预处理语句
        ps.close();
    }
}
/*
长连接:
    当连接数据的一个连接, 长时间没有通讯的的时候, 连接会被数据库自动关闭
    比如: mysql 是当超过 8 个小时空闲, 则会自动关闭
    
 解决:
 1. 每隔一段时间, 与服务器进行一次通讯
    select 1;
 2. 每次使用连接连接之前, 判断连接释放关闭. 如果已经关闭, 重新获取新的链接.
 
 3. 使用 jdbc 连接池
    德鲁伊
 */